package org.event.core.service.producer.impl;

import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.event.core.entity.DeviceWarnInfo;
import org.event.core.enums.KafkaSendEnum;
import org.event.core.service.producer.KafkaProduceService;
import org.event.core.util.KafkaTopicUtil;
import org.event.core.util.SpringContextUtil;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @author zhangjm
 * @date 2023年11月29日 11:33
 */
@Slf4j
@Service("kafkaProducerService")
public class KafkaProducerServiceImpl implements KafkaProduceService {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${topic.name}")
    private String topicName;


    @Override
    public boolean sendWarnMessage(DeviceWarnInfo deviceWarnInfo) {
        boolean res = false;
        Map<String, Object> reportData = new HashMap<>();
        reportData.put("command", deviceWarnInfo.getCommand());
        reportData.put("data", deviceWarnInfo);
        //判断是否存在当前主题
        Set<String> topics = KafkaTopicUtil.getTopics(bootstrapServers);
        if (!topics.contains(KafkaSendEnum.ALARM_WARN_PUSH.getTopic())) {
            if (!KafkaTopicUtil.createTopics(bootstrapServers, topicName, 1, (short) 1)) {
                log.info("topic创建失败，消息发送不成功！");
                return res;
            }
        }

        KafkaTemplate kafkaTemplate = (KafkaTemplate) SpringContextUtil.getBean("kafkaTemplate");
        ListenableFuture send = kafkaTemplate.sendDefault(topicName, JSONArray.toJSONString(reportData));
        send.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                log.error(ex.getMessage() + "发送失败！原因：" + ex.getCause());
                System.out.println("发送失败!");
            }

            @Override
            public void onSuccess(Object result) {
                log.info("消息发送成功" + result.toString());
                System.out.println("发送成功！");
            }
        });
        return res;

    }
}
